package mysql;

import com.google.common.collect.Maps;
import org.apache.storm.jdbc.common.Column;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
import org.apache.storm.jdbc.common.JdbcClient;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class WordCountBolt extends BaseRichBolt {
    private  Map<String, Integer> wordMap;
    private JdbcClient jdbcClient;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        wordMap = new HashMap<>();
        Map<String, Object> mysqlConfig = Maps.newHashMap();
        mysqlConfig.put("dataSourceClassName", "com.mysql.cj.jdbc.MysqlDataSource");
        mysqlConfig.put("dataSource.url", "jdbc:mysql://liu/mydb");
        mysqlConfig.put("dataSource.user","root");
        mysqlConfig.put("dataSource.password","666666");

        ConnectionProvider connectionProvider = new HikariCPConnectionProvider(mysqlConfig);
        // 初始化数据连接池
        connectionProvider.prepare();
        jdbcClient = new JdbcClient(connectionProvider, 30);
    }

    @Override
    public void execute(Tuple input) {
        String word = input.getStringByField("word");
        int count = 1;
        if (wordMap.containsKey(word)) {
            count = wordMap.get(word) + 1;
        }

        wordMap.put(word, count);

        List<Column> columns = new ArrayList<>();
        columns.add(new Column("word", word, Types.VARCHAR)); /* Types 是 sql 中的一个枚举类 */
        List<List<Column>> select = jdbcClient.select("select * from wordcount where word = ?", columns);
        // 计算结果集中数据的条数
        long result = select.stream().count();
        String sql = null;
        if (result >= 1) {
            /*容易引起SQL注入，例如变量是"or id<>1" #   */
            sql = "update wordcount set count = " + wordMap.get(word) + " where word = '" + word + "'";
        } else {
            sql = "insert into wordcount values ('"+word+"', "+wordMap.get(word)+")";
        }
        jdbcClient.executeSql(sql);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}
